package com.heima.search.listener;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import com.heima.common.constants.ArticleConstants;
import com.heima.model.search.vos.SearchArticleVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Slf4j
@Component
public class SyncArticleListener {

    @Resource
    private ElasticsearchClient elasticsearchClient;
    private static final String INDEX_NAME = "app_info_article";

    @KafkaListener(topics = ArticleConstants.ARTICLE_ES_SYNC_TOPIC)
    public void onMessage(String message) {
        if (StrUtil.isEmpty(message)) {
            return;
        }
        try {
            // 将消息转为对象
            SearchArticleVo searchArticleVo = JSONUtil.toBean(message, SearchArticleVo.class);
            // 新增或更新索引数据
            elasticsearchClient.index(index -> index
                    .index(INDEX_NAME)
                    .id(searchArticleVo.getId().toString())
                    .document(searchArticleVo));
        } catch (Exception e) {
            log.error("数据写入ES失败！ message = {}", message, e);
        }
    }
}
